package com.lagou.rpc.consumer.server;

import com.lagou.rpc.consumer.handler.ClientHandler;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

import java.util.concurrent.*;

public class NettyClientServer {

    private NioEventLoopGroup eventExecutors;

    private Channel channel;

    private String ip;

    private int port;


    private ClientHandler clientHandler = new ClientHandler() ;

   private int cpuSize = Runtime.getRuntime().availableProcessors();

    private ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2,
            cpuSize,
            10,
            TimeUnit.SECONDS,
            new LinkedBlockingDeque<>(3),
            Executors.defaultThreadFactory(),
            new ThreadPoolExecutor.AbortPolicy());




    public NettyClientServer(String ip, int port) {
        this.ip = ip;
        this.port = port;
        init();
    }


    public void init(){
        try {
            eventExecutors = new NioEventLoopGroup();
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(eventExecutors)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS,3000)
                    .option(ChannelOption.SO_KEEPALIVE,true)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            ChannelPipeline pipeline = socketChannel.pipeline();
                            pipeline.addLast(new StringDecoder());
                            pipeline.addLast(new StringEncoder());
                            pipeline.addLast(clientHandler);
                        }
                    });
            channel = bootstrap.connect(ip, port).sync().channel();
        } catch (Exception e) {
            e.printStackTrace();

            if(channel != null){
                channel.close();
            }

            if(eventExecutors != null){
                eventExecutors.shutdownGracefully();
            }
        }


    }

    public void close(){
        if(channel != null){
            channel.close();
        }

        if(eventExecutors != null){
            eventExecutors.shutdownGracefully();
        }
    }

    public Object send(String msg) throws ExecutionException, InterruptedException {
        clientHandler.setMessage(msg);
        Future submit = threadPoolExecutor.submit(clientHandler);
        return submit.get();
    }
}
